package com.atguigu.flink0922.chapter08;

import com.atguigu.flink0922.bean.LoginEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/9 10:42
 */
public class Flink05_High_Project_Login {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        
        // 创建WatermarkStrategy
        WatermarkStrategy<LoginEvent> wms = WatermarkStrategy
            .<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(20))
            .withTimestampAssigner(new SerializableTimestampAssigner<LoginEvent>() {
                @Override
                public long extractTimestamp(LoginEvent element, long recordTimestamp) {
                    return element.getEventTime();
                }
            });
        env
            .readTextFile("input/LoginLog.csv")
            .map(line -> {
                String[] data = line.split(",");
                return new LoginEvent(Long.valueOf(data[0]),
                                      data[1],
                                      data[2],
                                      Long.parseLong(data[3]) * 1000L);
            })
            .assignTimestampsAndWatermarks(wms)
            .keyBy(LoginEvent::getUserId)
            .process(new KeyedProcessFunction<Long, LoginEvent, String>() {
                
                private ListState<Long> state;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    state = getRuntimeContext().getListState(new ListStateDescriptor<Long>("state", Long.class));
                }
                
                @Override
                public void processElement(LoginEvent input,
                                           Context ctx,
                                           Collector<String> out) throws Exception {
                    /*
                    两次失败不超过2s
                    1. 需要一个状态存两个失败的时间: ListState
                    2. 当列表的长度等于2的时候, 计算时间差是否在2秒
                      是: 检测到一个恶意登录
                      不是: 把第一个移除
                      
                    3. 如果碰到一个成功
                            清空状态
                    
                     */
                    if ("fail".equalsIgnoreCase(input.getEventType())) {
                        // 1. 把时间戳存入状态
                        state.add(input.getEventTime());
                        
                        final ArrayList<Long> tss = new ArrayList<>();
                        for (Long ts : state.get()) {
                            tss.add(ts);
                        }
                        tss.sort((o1, o2) -> (int) (o1 - o2));
                        
                        if (tss.size() == 2) { // 有两个失败
                            final long deltaTs = (tss.get(1) - tss.get(0)) / 1000;
                            if (deltaTs <= 2) {
                                out.collect(ctx.getCurrentKey() + " 在恶意登录");
                            }// 失败时间大于2秒, 把第一个失败去掉
                            tss.remove(0);
                            state.update(tss);
                        }
                    } else {
                        state.clear();
                    }
                    
                }
            })
            .print();
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
